package com.xqboss.common.core.mq;

import com.xqboss.common.config.properties.RocketMqConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.DependsOn;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.scheduling.annotation.Async;

import javax.annotation.PreDestroy;
import java.time.Duration;
import java.util.Collections;
import java.util.Objects;

@Configuration
@Slf4j
public abstract class MqListener implements ApplicationRunner {

    @Autowired
    private RocketMqConfig rocketMqConfig;
    private PushConsumer consumer;

    /**
     * 开启消费者监听服务
     *
     * @return
     */
    public PushConsumer listener(MqEnum mqEnum, Integer threadNum) throws Exception {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        SessionCredentialsProvider sessionCredentialsProvider =
                new StaticSessionCredentialsProvider(rocketMqConfig.getAccessKey(), rocketMqConfig.getSecretKey());
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(rocketMqConfig.getNameSrvAddr())
                .setCredentialProvider(sessionCredentialsProvider)
                .setRequestTimeout(Duration.ofSeconds(10))
                .build();
        PushConsumer consumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumptionThreadCount(threadNum)
                .setConsumerGroup(rocketMqConfig.getGroupName())
                .setSubscriptionExpressions(Collections.singletonMap(rocketMqConfig.getTopicPrefix() + mqEnum.getTopic(), new FilterExpression(mqEnum.getTag(), FilterExpressionType.TAG)))
                .setMessageListener(MqListener.this::doConsume)
                .build();
        log.info("开启 {} - {} - 启动成功 ---", mqEnum.getTopic(), mqEnum.getTag());
        return consumer;
    }

    public abstract MqEnum getEnum();

    public ConsumeResult doConsume(MessageView message) {
//        traceUtil.injectContext(message);
        log.info("Receive message - {} - {} - {}", getEnum().getMessage(), message.getTopic(), message.getMessageId());
        try {
            ConsumeResult consumeResult = solve(message);
            return consumeResult;
        } catch (Exception e) {
            log.error("{} - 消费失败 - e:{}", getEnum().getMessage(), e);
        }
        return ConsumeResult.SUCCESS;
    }

    /**
     * 执行埋点操作
     *
     * @param message
     */
    public abstract ConsumeResult solve(MessageView message);

    @Override
    public void run(ApplicationArguments args) throws Exception {
        try {
            this.consumer = listener(getEnum(), 20);
        } catch (Exception e) {
            log.error("{} - 消费者监听器启动失败 - {}", getEnum().getMessage(), e);
            throw e;
        }
    }

}
